package com.example.rxbyhand;

import com.example.rxbyhand.scheduler.Scheduler;

/**
 * Author by lp,on 2021/8/16/016,16:49.
 * 被观察者核心入口类
 * 也是框架的入口
 */
public abstract class Observable<T> implements ObservableSource<T> {

    @Override
    public void subscribe(Observer observer) {
        //和谁建立订阅？怎么建立？
        //为了保证扩展性，交给开发人员实现，这里只是提供抽象方法

        subscribeActual(observer);

    }

    //交给子类实现
    protected abstract void subscribeActual(Observer<T> observer);

    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        return new ObservableCreate<T>(source);
    }

    public <R> ObservableMap<T, R> map(Function<T, R> function) {
        return new ObservableMap<>(this, function);
    }

    public <R> ObservableFlatMap<T, R> flatMap(Function<T, ObservableSource<R>> function) {
        return new ObservableFlatMap<>(this, function);
    }

    public ObservableSubscribeOn<T> subscribeOn(Scheduler scheduler) {
        return new ObservableSubscribeOn<>(this, scheduler);
    }

    public ObservableObserveOn<T> observeOn(Scheduler scheduler) {
        return new ObservableObserveOn<>(this, scheduler);
    }

}
